package com.lagou.dw.interceptor;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Strings;
import org.apache.commons.codec.Charsets;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.interceptor.Interceptor;
import org.junit.Test;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class LogTypeInterceptor1 implements Interceptor {
    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {
        String eventbody = new String(event.getBody(), Charsets.UTF_8);
        Map<String, String> headerMap = event.getHeaders();
        String[] bodyArr = eventbody.split("\\s+");

        try {
            String jsonStr = bodyArr[6];
            if (Strings.isNullOrEmpty(jsonStr)) {
                return null;
            }

            String timestampStr = "";
            JSONObject jsonObject = JSON.parseObject(jsonStr);
            if (headerMap.getOrDefault("logtype", "").equals("start")){
                // 取启动日志的时间戳
                timestampStr = jsonObject.getJSONObject("app_active").getString("time");
            } else if (headerMap.getOrDefault("logtype","").equals("event")) {
                // 取事件日志第一条记录的时间戳
                JSONArray jsonArray = jsonObject.getJSONArray("lagou_event");
                if (jsonArray.size() > 0){
                    timestampStr = jsonArray.getJSONObject(0).getString("time");
                }
            }

            // 将 timestamp 转为 时间日期类型(格式:yyyy-MM-dd)
            long timeStamp = Long.parseLong(timestampStr);
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
            String date = formatter.format(LocalDateTime.ofInstant(Instant.ofEpochMilli(timeStamp), ZoneId.systemDefault()));
            headerMap.put("logtime", date);
            event.setHeaders(headerMap);
            // 设置event的body为有效json串
            event.setBody(jsonStr.getBytes());
        } catch (Exception e) {
            headerMap.put("logtime", "Unknown");
            event.setHeaders(headerMap);
        }
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        List<Event> events1 = new ArrayList<>();
        for(Event event : events) {
            Event outEvent = intercept(event);
            if (outEvent != null) {
                events.add(outEvent);
            }
        }
        return events1;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {
        @Override
        public Interceptor build() {
            return new LogTypeInterceptor();
        }
        @Override
        public void configure(Context context) {
        }
    }

    @Test
    public void test() {
        String str = "2020-08-20 11:56:09.947 [main] INFO  com.lagou.ecommerce.AppStart - {\"app_active\":{\"name\":\"app_active\",\"json\":{\"entry\":\"2\",\"action\":\"0\",\"error_code\":\"0\"},\"time\":1595470669202},\"attr\":{\"area\":\"邯郸\",\"uid\":\"2F10092A8000\",\"app_v\":\"1.1.7\",\"event_type\":\"common\",\"device_id\":\"1FB872-9A1008000\",\"os_type\":\"9.3\",\"channel\":\"RX\",\"language\":\"chinese\",\"brand\":\"Huawei-9\"}}";
        Event event = new SimpleEvent();
        event.setBody(str.getBytes(Charsets.UTF_8));
        CustomerInterceptor customerInterceptor = new CustomerInterceptor();
        Event intercept = customerInterceptor.intercept(event);
        Map<String, String> headers = intercept.getHeaders();
        System.out.println(JSON.toJSONString(headers));
    }
}
